package org.nbict.iot.trash;

import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.mongodb.common.QueryFilterCreator;
import org.apache.storm.mongodb.common.SimpleQueryFilterCreator;
import org.apache.storm.mongodb.common.mapper.MongoMapper;
import org.apache.storm.mongodb.common.mapper.SimpleMongoMapper;
import org.apache.storm.topology.TopologyBuilder;
import org.nbict.iot.PropertyUtil;
import org.nbict.iot.trash.mogo.MongoUpdateBolt;

/**
 * Created by songseven on 18/6/20.
 */
@Deprecated
public class RealtimeTopology {

    TopologyBuilder builder = new TopologyBuilder();

    public static void main(String[] args) throws InvalidTopologyException,
            AuthorizationException, AlreadyAliveException {

        TopologyBuilder builder = new TopologyBuilder();

        KafkaSpoutConfig.Builder<String, String> kafkaBuilder = new
                KafkaSpoutConfig.Builder("127.0.0.1:9092", StringDeserializer
                .class, StringDeserializer.class, "transit-5vvokv");
        //transit-5vvokv 需要监听多个

        kafkaBuilder.setGroupId("iot-car-group-realtime-test-3");
        KafkaSpoutConfig build = kafkaBuilder.build();
        KafkaSpout kafkaSpout = new KafkaSpout(build);

        //step1. 报文读取流
        builder.setSpout("kafkaSpout", kafkaSpout, 1);

        //step2. 报文解析分发
        builder.setBolt("dispatcher", new DispatcherRealPacketBolt(), 2)
                .shuffleGrouping("kafkaSpout");



        String database = PropertyUtil.getProperty("mongo.database");
        String username = PropertyUtil.getProperty("mongo.username");
        String password = PropertyUtil.getProperty("mongo.password");
        String adminname = PropertyUtil.getProperty("mongo.admin.name");
        String url = PropertyUtil.getProperty("mongo.url");

        //step3. 更新车辆实时各维度
        MongoMapper realUpdateMapper = new SimpleMongoMapper().withFields
                ("platform_id", "car_vin", "tag", "timestamp", "info");
        QueryFilterCreator realUpdateQueryCreator = new
                SimpleQueryFilterCreator().withField("platform_id").withField
                ("car_vin").withField("tag");
        MongoUpdateBolt realUpdateBolt = new MongoUpdateBolt(url, username,
                password, adminname, database, "car", realUpdateQueryCreator,
                realUpdateMapper).withUpsert(true);
        builder.setBolt("realtimeUpdate", realUpdateBolt, 2).shuffleGrouping
                ("dispatcher", "carRealtime");

        //step4. 更新地图车辆状态
        MongoMapper carStatusUpdateMapper = new SimpleMongoMapper()
                .withFields("car_vin", "timestamp", "car_status",
                        "charge_status");
        QueryFilterCreator carStatusUpdateQueryCreator = new
                SimpleQueryFilterCreator().withField("car_vin");
        MongoUpdateBolt carStatusUpdateBolt = new MongoUpdateBolt(url,
                username, password, adminname, database, "vehicle_map",
                carStatusUpdateQueryCreator, carStatusUpdateMapper)
                .withUpsert(true);
        builder.setBolt("carStatusUpdate", carStatusUpdateBolt, 1)
                .shuffleGrouping("dispatcher", "carInfoMap");

        //step5. 更新地理
        MongoMapper carGeoUpdateMapper = new SimpleMongoMapper().withFields
                ("car_vin", "timestamp", "lon", "lat");
        QueryFilterCreator carGeoUpdateQueryCreator = new
                SimpleQueryFilterCreator().withField("car_vin");
        MongoUpdateBolt carGeoUpdateBolt = new MongoUpdateBolt(url, username,
                password, adminname, database, "vehicle_map",
                carGeoUpdateQueryCreator, carGeoUpdateMapper).withUpsert(true);
        builder.setBolt("carGeoUpdate", carGeoUpdateBolt, 1).shuffleGrouping
                ("dispatcher", "carGeoMap");

        //step6. 更新报警
        MongoMapper carAlarmUpdateMapper = new SimpleMongoMapper().withFields
                ("car_vin", "timestamp", "alarm_status");
        QueryFilterCreator carAlarmUpdateQueryCreator = new
                SimpleQueryFilterCreator().withField("car_vin");
        MongoUpdateBolt carAlarmUpdateBolt = new MongoUpdateBolt(url,
                username, password, adminname, database, "vehicle_map",
                carAlarmUpdateQueryCreator, carAlarmUpdateMapper).withUpsert
                (true);
        builder.setBolt("carAlarmUpdate", carAlarmUpdateBolt, 1)
                .shuffleGrouping("dispatcher", "carAlarmMap");

        Config config = new Config();

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, builder
                    .createTopology());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("iot-car-realtime", config, builder
                    .createTopology());
        }
    }
}
